package com.bw.gmall.realtime.dwd.db.app;

import com.bw.gmall.realtime.common.base.BaseSQLApp;
import com.bw.gmall.realtime.common.constant.Constant;
import com.bw.gmall.realtime.common.util.SQLUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.TableResult;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

public class DwdTradeOrderCancelDetail extends BaseSQLApp {
    public static void main(String[] args) {
        new DwdTradeOrderCancelDetail().start(10015,4, Constant.TOPIC_DWD_TRADE_ORDER_CANCEL);
    }

    @Override
    public void handle(StreamExecutionEnvironment env, StreamTableEnvironment tEnv, String groupId) {

        // 1. 读取 topic_db 数据
        readOdsDb(tEnv,groupId);

        // 2. 读取 dwd 层下单事务事实表数据
        kafkacreate(tEnv);

        // 3. 从 topic_db 过滤出订单取消数据

        order_cancel(tEnv);

        // 4. 订单取消表和下单表进行 join
        Table table = canel_order(tEnv);
        // 5. 写出

        //创建kafka 结构
        create_ordercanel(tEnv);

        //写入kafka
        table.insertInto(Constant.TOPIC_DWD_TRADE_ORDER_CANCEL).execute();

    }

    private static void create_ordercanel(StreamTableEnvironment tEnv) {
        TableResult tableResult = tEnv.executeSql("create table " + Constant.TOPIC_DWD_TRADE_ORDER_CANCEL + "\n" +
                "(\n" +
                "id string,\n" +
                "order_id string,\n" +
                "user_id string,\n" +
                "sku_id string,\n" +
                "sku_name string,\n" +
                "province_id string,\n" +
                "activity_id string,\n" +
                "activity_rule_id string,\n" +
                "coupon_id string,\n" +
                "e_format string,\n" +
                "operate_time string,\n" +
                "sku_num string,\n" +
                "split_coupon_amount string,\n" +
                "split_total_amount string,\n" +
                "split_original_amount string,\n" +
                "ts bigint\n" +
                ")" + SQLUtil.getKafkaSinkSQL(Constant.TOPIC_DWD_TRADE_ORDER_CANCEL));
    }

    private static Table canel_order(StreamTableEnvironment tEnv) {
        return  tEnv.sqlQuery("select\n" +
                "od.id,\n" +
                "od.order_id,\n" +
                "od.user_id,\n" +
                "od.sku_id,\n" +
                "od.sku_name,\n" +
                "od.province_id,\n" +
                "od.activity_id,\n" +
                "od.activity_rule_id,\n" +
                "od.coupon_id,\n" +
                "date_format(oc.operate_time,'yyyy-MM-dd') order_cancel_date_id,\n" +
                "oc.operate_time,\n" +
                "od.sku_num,\n" +
                "od.split_coupon_amount,\n" +
                "od.split_total_amount,\n" +
                "od.split_original_amount,\n" +
                "oc.ts\n" +
                "from dwd_trade_order_detail od\n" +
                "join \n" +
                "order_cancel oc\n" +
                "on oc.id=od.order_id");
    }

    //订单取消
    private static void order_cancel(StreamTableEnvironment tEnv) {
        Table table = tEnv.sqlQuery("select \n" +
                "`data`['id'] id,\n" +
                "`data`['operate_time'] operate_time,\n" +
                "`ts` ts\n" +
                "from topic_db\n" +
                "where `database`='gmall' and `table`='order_info'\n" +
                "and type='update'\n" +
                "and `old`['order_status']='1001'\n" +
                "and `data`['order_status']='1003' ");

        tEnv.createTemporaryView("order_cancel",table);
    }

    //创建并读取 kafka主题数据
    private static void kafkacreate(StreamTableEnvironment tEnv) {
        TableResult tableResult = tEnv.executeSql("create table dwd_trade_order_detail(" +
                " id string,\n" +
                " order_id string,\n" +
                " sku_id string,\n" +
                " user_id string,\n" +
                " province_id string,\n" +
                " activity_id string,\n" +
                " activity_rule_id string,\n" +
                " coupon_id string,\n" +
                " sku_name string,\n" +
                " order_price string,\n" +
                " sku_num string,\n" +
                " create_time string,\n" +
                " split_total_amount string,\n" +
                " split_activity_amount string,\n" +
                " split_coupon_amount string,\n" +
                " split_original_amount string,\n" +
                " ts bigint\n" +
                ")" + SQLUtil.getKafkaSourceSQL(Constant.TOPIC_DWD_TRADE_ORDER_DETAIL,Constant.TOPIC_DWD_TRADE_ORDER_CANCEL));
    }
}
